package com.gy.spark.sparkstreaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Durations, StreamingContext}

object Operator_transform {
  def main(args: Array[String]): Unit = {

    val filterData = Array("?", "!")

    val conf = new SparkConf()
    val ssc = new StreamingContext(conf, Durations.seconds(5))

    val broadcastFilterData = ssc.sparkContext.broadcast(filterData)

    val ds = ssc.socketTextStream("localhost", 9999)
    val mapDs: DStream[(String, Int)] = ds.flatMap(_.split("")).map((_, 1))

    val resultDs: DStream[(String, (Int, Option[Boolean]))] = mapDs.transform {
//      println("driver running ...")
      rdd => {
        val fileRDD: RDD[(String, Boolean)] = ssc.sparkContext.parallelize(broadcastFilterData.value).map((_, true))
        val fileResultRDD: RDD[(String, (Int, Option[Boolean]))] = rdd.leftOuterJoin(fileRDD)
        fileResultRDD.filter(_._2._2.isEmpty)
      }
    }

    resultDs.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}

